package com.atguigu.ordertimeout_detect

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * 使用connect合并两条流
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.ordertimeout_detect
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/16 14:14
 */
object TxMatch {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val path1: String = TxMatch.getClass.getClassLoader.getResource("OrderLog.csv").getPath
    //第一条流
    val orderStream = env.readTextFile(path1)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          OrderEvent(dataArrays(0).toLong, dataArrays(1), dataArrays(2), dataArrays(3).toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(1)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })
      .filter(_.eventType == "pay")
      .keyBy(_.pyId)


    val path2: String = TxMatch.getClass.getClassLoader.getResource("ReceiptLog.csv").getPath
    //第二条流
    val receiptLogStream = env.readTextFile(path2)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          ReceiptEvent(dataArrays(0), dataArrays(1), dataArrays(2).toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ReceiptEvent](Time.seconds(1)) {
        override def extractTimestamp(element: ReceiptEvent): Long = element.eventTime * 1000L
      })
      .keyBy(_.tyId)


    //使用connect合并2条流
    val value: DataStream[(OrderEvent, ReceiptEvent)] = orderStream.connect(receiptLogStream)
      .process(new MyConnect())

    value.print("output")
    value.getSideOutput(new OutputTag[String]("msg")).print("sideOutput")

    env.execute("join job")
  }
}

class MyConnect() extends CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]{

  lazy val orderState: ValueState[OrderEvent] = getRuntimeContext.getState(new ValueStateDescriptor[OrderEvent]("order", classOf[OrderEvent]))

  lazy val receoptState: ValueState[ReceiptEvent] = getRuntimeContext.getState(new ValueStateDescriptor[ReceiptEvent]("receipt", classOf[ReceiptEvent]))

  override def processElement1(value: OrderEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    val receiptEvent: ReceiptEvent = receoptState.value()

    if (receiptEvent != null) {
      out.collect(value, receiptEvent)
      receoptState.clear()
    } else{
      orderState.update(value)
      //等待5秒
      ctx.timerService().registerEventTimeTimer(value.eventTime * 1000L + 5000L)
    }

  }

  override def processElement2(value: ReceiptEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    val orderEvent: OrderEvent = orderState.value()

    if (orderEvent != null) {
      out.collect(orderEvent, value)
      orderState.clear()
    } else {
      receoptState.update(value)
      //等待5秒
      ctx.timerService().registerEventTimeTimer(value.eventTime * 1000L + 3000L)
    }
  }

  override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    val orderEvent = orderState.value()
    if (orderState.value() != null){
      ctx.output(new OutputTag[String]("msg"), orderEvent.pyId + " no receipt")
    }

    val receoptEvent = receoptState.value()
    if (receoptState.value() != null){
      ctx.output(new OutputTag[String]("msg"), receoptEvent.tyId + " no pay")
    }

    orderState.clear()
    receoptState.clear()
  }
}